package com.example.latch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author chentao
 * @version 1.0
 * @description: TODO
 * @date 2025/3/23 下午7:36
 */
public class MySqlSyncLatch implements MySqlLatch{

    Logger logger = LoggerFactory.getLogger(getClass());

    private DataSource dataSource;

    private Connection connection;

    private PreparedStatement preparedStatement;

    private boolean connAutoCommit = false;

    //分布式锁的SQL
    private String sql = "select * from sys_lock where id = 2 for update";

    private volatile boolean running = true;

    //心跳线程池核心数
    private int corePoolSize = 1;

    //心跳线程池心跳频率 默认30秒
    private int timeInterval = 30;

    //心跳线程池心跳的单位 默认秒
    private TimeUnit timeUnit = TimeUnit.SECONDS;

    private MysqlLatchListener mysqlLatchListener;

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    public MySqlSyncLatch(DataSource dataSource, MysqlLatchListener mysqlLatchListener) {
        this.dataSource = dataSource;
        this.mysqlLatchListener = mysqlLatchListener;
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(this.corePoolSize);
    }

    public MySqlSyncLatch(DataSource dataSource, String sql, MysqlLatchListener mysqlLatchListener) {
        this.dataSource = dataSource;
        this.sql = sql;
        this.mysqlLatchListener = mysqlLatchListener;
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(this.corePoolSize);
    }

    public MySqlSyncLatch(DataSource dataSource, String sql, MysqlLatchListener mysqlLatchListener, int corePoolSize, int timeInterval) {
        this.dataSource = dataSource;
        this.sql = sql;
        this.mysqlLatchListener = mysqlLatchListener;
        this.corePoolSize = corePoolSize;
        this.timeInterval = timeInterval;
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(this.corePoolSize);
    }

    /****
     * 开启线程并开始选举操作
     * @throws Exception
     */
    @Override
    public void start() throws Exception {
        boolean lockAcquired = false;
        while (running) {
            try {
                connection = dataSource.getConnection();
                connAutoCommit = connection.getAutoCommit();
                connection.setAutoCommit(false);
                preparedStatement = connection.prepareStatement(sql);
                preparedStatement.execute();
                lockAcquired = true;
                break;
            } catch (Exception e) {
                closeAndClean();
            }
        }
        if (lockAcquired) {
            scheduledThreadPoolExecutor.scheduleAtFixedRate(new MySqlLatchSQLConnectionActiveThread(connection), 0, timeInterval, timeUnit);
            mysqlLatchListener.isLeader();
        } else {
            logger.error("Failed to acquire lock after multiple attempts.");
        }
    }

    /****
     * 关闭并回收资源
     */
    @Override
    public void closeAndClean() {
        if (preparedStatement != null) {
            try {
                preparedStatement.close();
            } catch (SQLException e) {
                logger.warn("preparedStatement close error: " + e.getMessage());
            }
        }
        if (connection != null) {
            try {
                connection.commit();
            } catch (SQLException e) {
                logger.warn("connection commit error: " + e.getMessage());
            }
        }
        try {
            if (connection != null) {
                connection.setAutoCommit(connAutoCommit);
            }
        } catch (SQLException e) {
            logger.warn("connection set autocommit error: " + e.getMessage());
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                logger.warn("connection close error: " + e.getMessage());
            }
        }
    }

    /****
     * 停止并回收线程
     * @throws Exception
     */
    @Override
    public void stop() {
        this.running = false;
        this.closeAndClean();
        this.scheduledThreadPoolExecutor.shutdownNow();
    }
}
